package com.atguigu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.atguigu.gmall.realtime.app.function.CustomerDeserialization;
import com.atguigu.gmall.realtime.app.function.DimSinkFunction;
import com.atguigu.gmall.realtime.app.function.TableProcessFunction;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;

//数据流 web/app -> nginx -> SprinBAoot -> mysql -> flinkapp -> kafka(ods) ->flinkapp -> kafka(dwd)/phoenix()di
//程序              mockDB   -> MySQL-》 FlinkCDC -> kafka (ZK) -> BaseApp -> Kafka/Phoenix（HBASE，zk，hdfs）

public class BaseDBApp {
    public static void main(String[] args) throws Exception {
//        1.获取流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

//        //2.Flink-CDC将读取binlog的位置信息以状态的方式保存在CK,如果想要做到断点续传,需要从Checkpoint或者Savepoint启动程序
//        //2.1 开启Checkpoint,每隔5秒钟做一次CK
//        env.enableCheckpointing(5000L);
////        设置超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(10000L);
////        设置最大当前ck数量
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
////        设置最小ck间隔
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);
//        //2.2 指定CK的一致性语义
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        //2.3 设置任务关闭的时候保留最后一次CK数据
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        //2.4 指定从CK自动重启策略
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 2000L));
//        //2.5 设置状态后端
//        env.setStateBackend(new FsStateBackend("hdfs://djcluster/flinkCDC"));
//        //2.6 设置访问HDFS的用户名
//        System.setProperty("HADOOP_USER_NAME", "atguigu");

//       2. 消费Kafka数据 ods_base_db 主题数据创建流

        String sourceTopic = "ods_base_db";
        String groupId = "base_db_app";

        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer(sourceTopic, groupId));

//        3.将每行数据转换为JSON对象并过滤（delete）本次需求不需要  主流 (如果担心有脏数据可以用process try catch之后 去捕获异常)

        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSON::parseObject)
                .filter(new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
//                        取出操作数据类型
                        String type = value.getString("type");
//                      三元运算符取反  过滤
                        return !"delete".equals(type);
                    }
                });

//        4.使用flink cdc 消费配置表 并处理成  广播流
        DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("hadoop101")
                .port(3306)
                .username("root")
                .password("000000")
                .databaseList("gmall-realtime")
//                注意要使用 库.表名 来指定
                .tableList("gmall-realtime.table_process")
                //最新开始读
                .startupOptions(StartupOptions.latest())
                .deserializer(new CustomerDeserialization())
                .build();
//      配置流
        DataStreamSource<String> tableProcessStrDS = env.addSource(sourceFunction);
        MapStateDescriptor<String, TableProcess> mapStateDescriptor = new MapStateDescriptor<String, TableProcess>("map-state", String.class, TableProcess.class);

        //      把流广播出来
        BroadcastStream<String> broadcastStream = tableProcessStrDS.broadcast(mapStateDescriptor);


//      5.连接主流和广播流
        BroadcastConnectedStream<JSONObject, String> connectedStream = jsonObjDS.connect(broadcastStream);
        jsonObjDS.print("jsonObjDS=>>>>>>>>>>>>>>>>>>>");

//      6.处理数据 广播流数据 ，主流数据（根据广播流数据进行处理）
        OutputTag<JSONObject> hbaseTag = new OutputTag<JSONObject>("hbase-tag") {
        };

        SingleOutputStreamOperator<JSONObject> kafka = connectedStream.process(new TableProcessFunction(hbaseTag, mapStateDescriptor));

//      7.提取kafka流数据 和Hbase 流数据
        DataStream<JSONObject> hbase = kafka.getSideOutput(hbaseTag);

//      8.将kafka数据写入kafka主题 将hbase数据写入 phoenix表
        kafka.print("kafka =>>>>>>>>>>>>>>>");
        hbase.print("hbase =>>>>>>>>>>>>>>>");

//            hbase.addSink(JdbcSink.sink()) 这种JDBC SINK方式不能用 因为没办法确定具体参数个数
        hbase.addSink(new DimSinkFunction());

//        调用的时候 明确了类型
        kafka.addSink(MyKafkaUtil.getKafkaProducer(new KafkaSerializationSchema<JSONObject>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(JSONObject element, @Nullable Long timestamp) {
                return new ProducerRecord<byte[], byte[]>(element.getString("sinkTable"),
                        element.getString("after").getBytes());
            }
        }));



//      9.启动任务
        env.execute("BaseDBApp");


    }
}
